package com.shujia.dws

import com.shujia.common.utils.{DateUtil, Geography, SparkMain}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}

object DwsSpacetimeCompanionMskDay extends SparkMain {

  override def run(spark:SparkSession): Unit = {

    import spark.implicits._

    // 从MySQL中加载确诊人员名单
    val confirmMdnsDF: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306/crm?useSSL=false")
      .option("dbtable", "crm.confirm_mdns")
      .option("user", "root")
      .option("password", "123456")
      .load()

    // 加载位置数据融合表
    val mergeDF: Dataset[Row] = spark.table("dwi.dwi_res_regn_mergelocation_msk_d").where($"day_id" === dayId)

    // 将位置数据融合表中的位置记录分为两类：1、确诊人员的行动轨迹 2、正常人员的行动轨迹
    val confirmPosDF: DataFrame = mergeDF.join(confirmMdnsDF, List("mdn"), "inner")
      .select(
        $"mdn" as "c_mdn"
        , $"start_date" as "c_start_date"
        , $"county_id"
        , $"longi" as "c_longi"
        , $"lati" as "c_lati"
      )

    // 直接进行过滤 将正常人员的行动轨迹过滤出来
    val mdnsArr: Array[String] = confirmMdnsDF.map(row => row.getAs[String]("mdn")).collect()

    val normalPosDF: DataFrame = mergeDF.filter(row => !mdnsArr.contains(row.getAs[String]("mdn")))

    // 将 正常人员的行动轨迹 同 确诊人员的行动轨迹 对比 找到时空伴随者
    /**
     * 时空伴随者:
     * 1、距离小于100m
     * 2、时间前后不超过1小时
     */

    /**
     * 优化：(尽可能减少比较次数)
     * 1、当正常人员的行动轨迹同确诊人员的行动轨迹对比时，只要一旦产生了时空交集，就可以跳出当前循环，取后续的正常人员的行动轨迹进行对比
     * 2、将同一个正常人的行动轨迹分到同一组中，再和确诊人员的行动轨迹进行对比，只要一旦产生了时空交集，就可以取后续的正常人员的行动轨迹进行对比
     *
     * 3、使用区县id进行关联，大大减少比较次数，进而减少运行时间，提高效率
     */

    val spacetimeCompanionPosDF: DataFrame = normalPosDF
      .join(confirmPosDF.hint("broadcast"), List("county_id"), "inner")
      // 计算时间差
      .withColumn("diff_time", diffTime($"start_date", $"c_start_date"))
      .where($"diff_time" < 60 * 60)
      // 计算距离差
      .withColumn("distance", calculateLengthWithLongiLati($"longi", $"lati", $"c_longi", $"c_lati"))
      .where($"distance" < 100)
      // 整理数据
      .select(
        $"mdn"
        , $"start_date"
        , $"end_date"
        , $"county_id"
        , $"longi"
        , $"lati"
        , $"bsid"
        , $"grid_id"
        , $"c_mdn"
        , $"c_start_date"
        , $"c_longi"
        , $"c_lati"
      )


    spacetimeCompanionPosDF
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dws/dws_spacetime_companion_msk_d/day_id=$dayId")

    spark.sql(
      s"""
        |alter table dws.dws_spacetime_companion_msk_d add if not exists partition(day_id='$dayId')
        |""".stripMargin)


    // 提取时空伴随者的手机号去重后进行保存
    spacetimeCompanionPosDF
      .select($"mdn")
      .distinct()
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dws/dws_spacetime_companion_mdn_msk_d/day_id=$dayId")

    spark.sql(
      s"""
         |alter table dws.dws_spacetime_companion_mdn_msk_d add if not exists partition(day_id='$dayId')
         |""".stripMargin)

  }

  /**
   * 提交命令：
   * spark-submit --master yarn-client --conf spark.sql.shuffle.partitions=10 --class com.shujia.dws.DwsSpacetimeCompanionMskDay --jars common-1.0.jar dws-1.0.jar 20230105
   */
}
